/*
 * thread.c
 *
 * Copyright (C) 2011 - Dr.NP
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

/**
 * @version thread.c 1.0.0
 * @package libbspcore
 * @author Dr.NP <np@bsgroup.org>
 * @update 03/21/2011
 */

/**
 * Worker thread handler
 *
 * === CHANGELOG ===
 * [03/21/2011] - Creation
 */

#include "bsp.h"

#include "bsp_socket.h"
#include "bsp_timer.h"
#include "bsp_thread.h"
#include "bsp_utils.h"

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <err.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/epoll.h>

#ifdef HAVE_EVENTFD
#include <sys/eventfd.h>
#endif

#include <pthread.h>

struct worker_t *worker_list;
size_t worker_list_size;
int worker_total;
volatile int curr;

void * worker_init(void *);
int do_timer(int);
int next_worker();

int thread_init(int nworkers)
{
    if (nworkers > MAX_THREADS - 1)
    {
        nworkers = MAX_THREADS - 1;
    }

    worker_list = ensure_list_space(NULL, sizeof(struct worker_t), &worker_list_size, MAX_THREADS);
    if (!worker_list)
    {
        err(RTN_ERROR_MEMORY_ALLOC, "Thread list initialize error!!!");
    }

    memset(worker_list, 0, sizeof(struct worker_t) * MAX_THREADS);
    curr = 1;

    int i, ret;

    for (i = 1; i < nworkers + 1; i ++)
    {
        worker_list[i].id = i;
        worker_list[i].epoll_fd = epoll_create(EPOLL_WAIT_CONNS);

        set_fd(worker_list[i].epoll_fd, i, FD_TYPE_EPOLL);
#ifdef HAVE_EVENTFD
        // Eventfd 
        ret = eventfd(0, EFD_NONBLOCK);
        if (ret < 0)
        {
            err(RTN_ERROR_IO_EVENT, "Thread event pipe error!!!");
        }

        worker_list[i].notify_event_fd = ret;
        set_fd(ret, i, FD_TYPE_EVENT);
#else
        // Pipe
        int fds[2];

        if (pipe(fps))
        {
            err(RTN_ERROR_IO_PIPE, "Thread event pipe error!!!");
        }

        worker_list[i].notify_recv_fd = fds[0];
        worker_list[i].notify_send_fd = fds[1];

        set_fd(fds[0], i, FD_TYPE_PIPE);
        set_fd(fds[1], i, FD_TYPE_PIPE);
#endif
        pthread_attr_init(&worker_list[i].attr);
        ret = pthread_create(&worker_list[i].tid, &worker_list[i].attr, worker_init, &worker_list[i]);

        worker_total ++;
    }

    debug_info("%d workers initialized\n", nworkers);

    return RTN_SUCCESS;
}

void * worker_init(void *arg)
{
    struct worker_t *w = (struct worker_t *) arg;
    struct epoll_event ev, events[EPOLL_WAIT_CONNS];
    struct fd_info_t *f;
    struct client_t *c;
    struct connector_t *n;
    struct timer_t *t;
    int nfds, i, pfd, what, task_id, task;
    u_int64_t sig;

#ifdef HAVE_EVENTFD
    ev.data.fd = w->notify_event_fd;
    ev.events = EPOLLIN;
    (void) epoll_ctl(w->epoll_fd, EPOLL_CTL_ADD, w->notify_event_fd, &ev);
#else
    ev.data.fd = w->notify_recv_fd;
    ev.events = EPOLLIN;
    (void) epoll_ctl(w->epoll_fd, EPOLL_CTL_ADD, w->notify_recv_fd, &ev);
#endif

    while (1)
    {
        memset(events, 0, sizeof(struct epoll_event) * EPOLL_WAIT_CONNS);
        nfds = epoll_wait(w->epoll_fd, events, EPOLL_WAIT_CONNS, -1);

        for (i = 0; i < nfds; i ++)
        {
            pfd = events[i].data.fd;
            what = events[i].events;

            if (pfd < 0)
            {
                continue;
            }

#ifdef HAVE_EVENTFD
            if (pfd != w->notify_event_fd)
#else
            if (pfd != w->notify_recv_fd)
#endif
            {
                f = get_fd(pfd);
                if (!f)
                {
                    continue;
                }
                
                switch (f->type)
                {
                    case FD_TYPE_TIMER : 
                        if (!what & EPOLLIN)
                        {
                            continue;
                        }
                        
                        while (sizeof(u_int64_t) == read(pfd, &sig, sizeof(u_int64_t)))
                        {
                            do_timer(f->id);
                        }
                        
                        break;

                    case FD_TYPE_CLIENT : 
                        if (what & EPOLLIN)
                        {
                            read_client(f->id);
                        }

                        if (what & EPOLLOUT)
                        {
                            send_client(f->id);
                        }
                        
                        break;

                    case FD_TYPE_CONNECTOR : 
                        if (what & EPOLLIN)
                        {
                            read_connector(f->id);
                        }

                        if (what & EPOLLOUT)
                        {
                            send_connector(f->id);
                        }
                        
                        break;

                    default : 
                        continue;
                }
            }

            else
            {
                while (sizeof(u_int64_t) == read(pfd, &sig, sizeof(u_int64_t)))
                {
                    task_id = sig & 0xFFFFFFFF;
                    task = (sig >> 32) & 0xFFFFFFFF;
                    f = get_fd(task_id);

                    switch (task)
                    {
                        case DISPATCH_ADD_TIMER : 
                            if (f)
                            {
                                if (f->type == FD_TYPE_TIMER)
                                {
                                    t = get_timer(f->id);
                                    if (t)
                                    {
                                        (void) epoll_ctl(w->epoll_fd, EPOLL_CTL_ADD, task_id, &t->ev);
                                    }
                                }
                            }
                            
                            break;

                        case DISPATCH_ADD_CLIENT : 
                            if (f)
                            {
                                if (f->type == FD_TYPE_CLIENT)
                                {
                                    c = get_client(f->id);
                                    if (c)
                                    {
                                        (void) epoll_ctl(w->epoll_fd, EPOLL_CTL_ADD, c->fd, &c->ev);
                                    }
                                }
                            }
                            
                            break;

                        case DISPATCH_ADD_CONNECTOR : 
                            if (f)
                            {
                                if (f->type == FD_TYPE_CONNECTOR)
                                {
                                    n = get_connector(f->id);
                                    if (n)
                                    {
                                        (void) epoll_ctl(w->epoll_fd, EPOLL_CTL_ADD, n->fd, &n->ev);
                                    }
                                }
                            }
                                    
                            break;

                        case DISPATCH_DEL : 
                            (void) epoll_ctl(w->epoll_fd, EPOLL_CTL_DEL, task_id, NULL);
                            
                            break;

                        case 0 : 
                        default : 
                            // Just touch
                            continue;
                    }
                }
            }
        }
    }
}

int next_worker()
{
    while (1)
    {
        if (++ curr >= worker_total + 1)
        {
            curr = 1;
        }

        if (worker_list[curr].id == curr)
        {
            return curr;
        }
    }

    return RTN_ERROR_FATAL;
}

struct worker_t * get_worker(int worker_id)
{
    if (worker_id < 1 || worker_id >= worker_total + 1)
    {
        return (struct worker_t *) NULL;
    }

    return &worker_list[worker_id];
}

void dispatch_timer(int timer_id)
{
    u_int64_t sig;
    int wid;
    struct timer_t *t = get_timer(timer_id);

    if (!t)
    {
        return;
    }

    wid = next_worker();
    sig = ((u_int64_t) DISPATCH_ADD_TIMER << 32 | (u_int64_t) t->fd);
#ifdef HAVE_EVENTFD
    write(worker_list[wid].notify_event_fd, &sig, sizeof(u_int64_t));
#else
    write(worker_list[wid].notify_send_fd, &sig, sizeof(u_int64_t));
#endif
    t->worker_id = wid;

    return;
}

void dispatch_client(int client_id)
{
    u_int64_t sig;
    int wid;
    struct client_t *c = get_client(client_id);

    if (!c)
    {
        return;
    }

    wid = next_worker();
    sig = ((u_int64_t) DISPATCH_ADD_CLIENT << 32 | (u_int64_t) c->fd);
#ifdef HAVE_EVENTFD
    write(worker_list[wid].notify_event_fd, &sig, sizeof(u_int64_t));
#else
    write(worker_list[wid].notify_send_fd, &sig, sizeof(u_int64_t));
#endif
    c->worker_id = wid;

    return;
}

void dispatch_connector(int connector_id)
{
    u_int64_t sig;
    int wid;
    struct connector_t *n = get_connector(connector_id);

    if (!n)
    {
        return;
    }

    wid = next_worker();
    sig = ((u_int64_t) DISPATCH_ADD_CONNECTOR << 32 | (u_int64_t) n->fd);
#ifdef HAVE_EVENTFD
    write(worker_list[wid].notify_event_fd, &sig, sizeof(u_int64_t));
#else
    write(worker_list[wid].notify_send_fd, &sig, sizeof(u_int64_t));
#endif
    n->worker_id = wid;

    return;
}

void remove_from_worker(int fd, int wid)
{
    if (wid < 1 || wid > worker_total)
    {
        return;
    }

    u_int64_t sig = ((u_int64_t) DISPATCH_DEL << 32 | (u_int64_t) fd);
#ifdef HAVE_EVENTFD
    write(worker_list[wid].notify_event_fd, &sig, sizeof(u_int64_t));
#else
    write(worker_list[wid].notify_send_fd, &sig, sizeof(u_int64_t));
#endif

    return;
}

void touch_worker(int wid)
{
    if (wid < 1 || wid > worker_total)
    {
        return;
    }

    u_int64_t sig = 0;
#ifdef HAVE_EVENTFD
    write(worker_list[wid].notify_event_fd, &sig, sizeof(u_int64_t));
#else
    write(worker_list[wid].notify_send_fd, &sig, sizeof(u_int64_t));
#endif

    return;
}
